package com.atguigu.day05;

import com.atguigu.utils.ClickEvent;
import com.atguigu.utils.ClickSource;
import com.atguigu.utils.UrlCountPerWindow;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

// 全窗口聚合函数的实现
// ./home: {
//     window-1: [...],
//     window-2: [...]
// }
public class Example1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .keyBy(r -> r.url)
                .process(new FakeWindow(5000L))
                .print();

        env.execute();
    }

    public static class FakeWindow extends KeyedProcessFunction<String, ClickEvent, UrlCountPerWindow> {
        private long windowSize;

        public FakeWindow(long windowSize) {
            this.windowSize = windowSize;
        }

        // key: WindowStartTime
        // value: [窗口中的所有元素]
        private MapState<Long, List<ClickEvent>> mapState;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            mapState = getRuntimeContext().getMapState(
                    new MapStateDescriptor<Long, List<ClickEvent>>(
                            "map-state",
                            Types.LONG,
                            Types.LIST(Types.POJO(ClickEvent.class))
                    )
            );
        }

        @Override
        public void processElement(ClickEvent value, Context ctx, Collector<UrlCountPerWindow> out) throws Exception {
            // 数据到达的机器时间
            long currTs = ctx.timerService().currentProcessingTime();
            // 如何计算currTs所属的窗口的开始时间？
            // 例如currTs是17，那么窗口开始时间是
            long windowStartTime = currTs - currTs % windowSize;
            long windowEndTime  = windowStartTime + windowSize;

            // 将数据分配到窗口中
            if (!mapState.contains(windowStartTime)) {
                // 说明窗口的第一条数据来了
                ArrayList<ClickEvent> elements = new ArrayList<>();
                elements.add(value);
                mapState.put(windowStartTime, elements);
            } else {
                List<ClickEvent> elements = mapState.get(windowStartTime);
                elements.add(value);
                mapState.put(windowStartTime, elements);
            }

            // 当机器时间到达了窗口结束时间-1毫秒时，触发窗口的闭合计算
            ctx.timerService().registerProcessingTimeTimer(windowEndTime - 1L);
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<UrlCountPerWindow> out) throws Exception {
            super.onTimer(timestamp, ctx, out);
            // 聚合计算
            long windowEndTime = timestamp + 1L;
            long windowStartTime = windowEndTime - windowSize;
            String url = ctx.getCurrentKey();
            long count = mapState.get(windowStartTime).size();
            out.collect(new UrlCountPerWindow(url, count, windowStartTime, windowEndTime));
            // 窗口触发计算以后，销毁窗口
            mapState.remove(windowStartTime);
        }
    }
}
// {
//     "key-1": {window-1: [....], window-2: [....]},
//     "key-2": {window-1: [....], window-2: [....]}
// }